package org.hbhk.hms.mq.tx.provider;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.selector.SelectMessageQueueByHash;
import org.apache.rocketmq.common.message.Message;
import org.hbhk.hms.mq.tx.mq.JsonUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.datasource.DataSourceUtils;
import org.springframework.jdbc.support.JdbcUtils;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;

import com.google.common.base.Function;
import com.google.common.collect.FluentIterable;
import com.google.common.collect.Lists;

public class DefaultMqRecordService implements IMqRecordService, InitializingBean {

	@Autowired
	private JdbcTemplate jdbcTemplate;

	private String cols = "tid,body,msg_key,err_msg,create_time,topic,status,retries";

	private ScheduledExecutorService service = Executors.newScheduledThreadPool(10);

	private Logger log = LoggerFactory.getLogger(getClass());

	@Override
	public void insert(MqRecord mqRecord) {
		String table = mqRecord.getTable();
		if (StringUtils.isNotEmpty(table)) {
			String err = mqRecord.getErrMsg();
			int status = 1;
			if (StringUtils.isNotEmpty(err)) {
				status = 0;
			}
			String sql = "insert into " + mqRecord.getTable() + "(" + cols + ") values(?,?,?,?,?,?," + status + ",1)";
			mqRecord.setCreateTime(new Date());
			Object[] params = { mqRecord.getId(), mqRecord.getRequest(), mqRecord.getShard(), mqRecord.getErrMsg(),
					mqRecord.getCreateTime(), mqRecord.getTopic() };
			jdbcTemplate.update(sql, params);
		}

	}

	@Override
	public void update(MqRecord mqRecord) {
		String err_msg = mqRecord.getErrMsg();
		if (StringUtils.isEmpty(err_msg)) {
			String sql = "update  " + mqRecord.getTable()
					+ " set status=?, modify_time=?,retries = ? where tid = ? and msg_key =?";
			Object[] params = { mqRecord.getStatus(), new Date(), mqRecord.getRetries(), mqRecord.getId(),
					mqRecord.getShard() };
			jdbcTemplate.update(sql, params);
		} else {
			String sql = "update  " + mqRecord.getTable()
					+ " set status=?,err_msg=?, modify_time=?,retries = ? where tid = ? and msg_key =?";
			Object[] params = { mqRecord.getStatus(), mqRecord.getErrMsg(), new Date(), mqRecord.getRetries(),
					mqRecord.getId(), mqRecord.getShard() };
			jdbcTemplate.update(sql, params);

		}

	}

	private final static SelectMessageQueueByHash SELECT_MESSAGE_QUEUE_BY_KEY = new SelectMessageQueueByHash();

	public SendResult sendMessage(MqRequest mqRequest) {

		try {
			String topic = mqRequest.getTopic();
			String key = mqRequest.getMsgKey();
			String mqMessageStr = mqRequest.getBody();
			String tags = mqRequest.getTags();
			if (StringUtils.isEmpty(tags)) {
				tags = "*";
			}
			String mqName = mqRequest.getMqName();
			DefaultMQProducer template = DefaultMessageSender.mqProducerCache.get(mqName);
			Message msg = new Message(topic, tags, key, mqMessageStr.getBytes());
			SendResult sendResult = template.send(msg, SELECT_MESSAGE_QUEUE_BY_KEY, key);
			if (log.isDebugEnabled()) {
				String re = sendResult.toString();
				log.debug("rocketMQ生产消息成功," + re);
			}
			return sendResult;
		} catch (Exception e) {
			log.error("mq record send failed", e);
			String err = ExceptionUtils.getMessage(e);
			throw new RuntimeException(err);
		}
	}

	@Override
	public void delete(MqRecord mqRecord) {

	}

	@Override
	public void afterPropertiesSet() throws Exception {
		String mqtxRetry = System.getProperty("mqtx.retry");
		if (StringUtils.isNotEmpty(mqtxRetry) && mqtxRetry.equals("on")) {
			service.scheduleAtFixedRate(new Runnable() {
				@Override
				public void run() {
					autoSendMsg();
				}
			}, 0, 5, TimeUnit.SECONDS);
		}
	}

	@Override
	public void autoSendMsg(String job) {
		if ("1".equals(job)) {
			autoSendMsg();
		}
	}

	private void autoSendMsg() {
		Set<String> mqTxtabs = DefaultMessageSender.mqTxtabs;
		for (String table : mqTxtabs) {
			try {
				log.debug("开始扫描：" + table);
				String sql = "select tid,body,retries,msg_key   from " + table
						+ " where  status = 0 and retries < 5 and create_time >= ? limit 50";
				Date qdate = DateUtils.addHours(new Date(), -1);
				Object[] params = { qdate };
				List<Map<String, Object>> requests = jdbcTemplate.queryForList(sql, params);
				if (!CollectionUtils.isEmpty(requests)) {
					log.info("本次扫描重试数据:" + requests.size());
					for (Map<String, Object> request : requests) {
						String json = String.valueOf(request.get("body"));
						log.error("retry msg :" + json);
						if (StringUtils.isNotEmpty(json)) {
							try {
								MqRequest mqRequest = JsonUtil.jsonToObject(json, MqRequest.class);
								sendMessage(mqRequest);
								MqRecord mqRecord = new MqRecord();
								mqRecord.setId((String.valueOf(request.get("tid"))));
								mqRecord.setShard((String.valueOf(request.get("msg_key"))));
								mqRecord.setStatus(1);
								mqRecord.setTable(table);
								mqRecord.setRetries(Integer.parseInt(String.valueOf(request.get("retries"))) + 1);
								// 更新mq事务表
								update(mqRecord);
							} catch (Exception e) {
								log.error("retry msg error", e);
								// 更新mq事务表
								MqRecord mqRecord = new MqRecord();
								mqRecord.setId((String.valueOf(request.get("tid"))));
								mqRecord.setShard((String.valueOf(request.get("shard"))));
								mqRecord.setStatus(0);
								mqRecord.setTable(table);
								mqRecord.setErrMsg(ExceptionUtils.getStackTrace(e));
								mqRecord.setRetries(Integer.parseInt(String.valueOf(request.get("retries"))) + 1);
								update(mqRecord);
							}
						}

					}

				}

			} catch (Exception e) {
				log.error("加载服务管理信息错误", e);
			}
		}
	}

	private void executeInsert(Connection conn, String recordTable, List<MqRecord> records) {

		PreparedStatement ps = null;
		// private String cols =
		// "tid,request,shard,err_msg,create_time,topic,status,retries";

		StringBuilder batchSql = new StringBuilder("INSERT INTO ");
		batchSql.append(recordTable).append("(" + cols + ") VALUES ")
				.append(StringUtils.repeat("(?,?,?,?,?,?,?,?),", records.size()));

		Object[] parameters = FluentIterable.from(records)
				.transformAndConcat(new Function<MqRecord, Iterable<Object>>() {
					@Override
					public Iterable<Object> apply(MqRecord mqRecord) {
						return Lists.newArrayList(mqRecord.getId(), mqRecord.getRequest(), mqRecord.getShard(),
								mqRecord.getErrMsg(), mqRecord.getCreateTime(), mqRecord.getTopic(),
								mqRecord.getStatus(), mqRecord.getRetries());
					}
				}).toArray(Object.class);

		try {
			ps = conn.prepareStatement(batchSql.substring(0, batchSql.length() - 1));
			for (int i = 0; i < parameters.length; i++) {
				ps.setObject(i + 1, parameters[i]);
			}
			int count = ps.executeUpdate();
			Assert.isTrue(count == records.size(), "mq record insert failed");
		} catch (SQLException e) {
			JdbcUtils.closeStatement(ps);
			DataSourceUtils.releaseConnection(conn, jdbcTemplate.getDataSource());
			throw new RuntimeException("mq record insert failed", e);
		} finally {
			JdbcUtils.closeStatement(ps);
			DataSourceUtils.releaseConnection(conn, jdbcTemplate.getDataSource());

		}
	}

	@Override
	public void insertBatch(List<MqRecord> mqRecords) {
		Map<String, List<MqRecord>> splitList = new HashMap<String, List<MqRecord>>();
		for (MqRecord mqRecord : mqRecords) {
			String tab = mqRecord.getTable();
			if (splitList.containsKey(tab)) {
				splitList.get(tab).add(mqRecord);
			} else {
				List<MqRecord> lists = new ArrayList<MqRecord>();
				lists.add(mqRecord);
				splitList.put(tab, lists);
			}
		}
		Set<String> keySet = splitList.keySet();
		for (String recordTable : keySet) {
			try {
				executeInsert(jdbcTemplate.getDataSource().getConnection(), recordTable, splitList.get(recordTable));
			} catch (SQLException e) {
			}
		}
	}

}
